package com.huawei.codearts.service;

import com.huawei.codearts.dao.RedisDao;
import com.huawei.codearts.tools.utils.JsonUtils;
import io.etcd.jetcd.ByteSequence;
import io.etcd.jetcd.Client;
import io.etcd.jetcd.KeyValue;
import io.etcd.jetcd.Watch;
import io.etcd.jetcd.options.GetOption;
import io.etcd.jetcd.watch.WatchEvent;
import io.etcd.jetcd.watch.WatchResponse;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.nio.charset.Charset;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.Map;
import java.util.Objects;
import java.util.Scanner;
import java.util.concurrent.ConcurrentHashMap;

import static java.nio.charset.StandardCharsets.UTF_8;

@Slf4j
@Service
public class AsyncService {

    @Autowired
    private RedisDao redisClient;

    @Autowired
    private Client etcdClient;

    @Async("myAsyncThreadPool")
    public void executeTask() throws InterruptedException {
        // 这里是你的任务逻辑
        System.out.println("异步任务执行中: " + Thread.currentThread().getName());
        Thread.sleep(3000);
        System.out.println("异步任务执行结束: " + Thread.currentThread().getName());
    }

    @Async("myAsyncThreadPool")
    public void deferQueueTask() throws InterruptedException {
        System.out.println("deferQueueTask is running..." + Thread.currentThread().getName());
        // 延迟队列key 可根据有续集合的分数处理
        String deferZset = "task::queue::defer";
        // 消息队列key
        String queueList = "task::queue";
        while (true) {
            long minScore = 0L;
            long maxScore = Instant.now().getEpochSecond();
            long number = redisClient.zsetTransferList(deferZset, minScore, maxScore, queueList);
            if (number <= 0) {
                LocalDateTime now = LocalDateTime.now();
                System.out.println("deferQueueTask redis延迟队列等待: " + Thread.currentThread().getName() + ", time: " + now.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
                Thread.sleep(5000);
            }
        }
    }

    /**
     * redis 队列演示
     */
    @Async("myAsyncThreadPool")
    public void queueTask() throws IOException, InterruptedException {
        System.out.println("queueTask is running..." + Thread.currentThread().getName());
        // 消息队列key
        String queueList = "task::queue";
        // 备份数据的消息队列key
        String queueDestination = "task::queue::destination";

        // 处理备份队列里是否有数据
        if (redisClient.llen(queueDestination) > 0) {
            redisClient.listTransferList(queueDestination, queueList);
        }

        // 开始监听队列
        String element = null;
        while (true) {
            String ack = null;
            try {
                element = redisClient.brpoplpush(queueList, queueDestination, 0);
                LocalDateTime now = LocalDateTime.now();
                String time = now.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
                log.info("queueTask [start] item: {}, time: {}, threadId: {}",element, time, Thread.currentThread().getName());
                // 业务处理 把json转换成map
                Map<String, Object> dataMap = JsonUtils.jsonToMap(element);
                // 验证消息 唯一消息ID messageId 是否被处理过了
                // 如果消息处理成功 保存 messageId 供以后验证 检查此消息是否消费过
                // messageId 存在则表明此消息已经被消费过
                // 多个服务时 此处可以加分布式锁以保证不会重复消费消息 例如：以messageId为粒度的锁
                // 再进一步可能需要事务加持 要么全部成功 要么都不成功
                if (dataMap.get("messageId") != "") {
                    // 模拟逻辑处理
                    Thread.sleep(3000);
                    LocalDateTime nowEnd = LocalDateTime.now();
                    String timeEnd = nowEnd.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
                    log.info("queueTask [end] item: {}, time: {}, threadId: {}",element, timeEnd, Thread.currentThread().getName());
                    // 处理完成设置状态
                    ack = "ok";
                }
                // 逻辑处理完成或没有消息唯一ID，删除备份队列数据
                if (Objects.equals(ack, "ok") || dataMap.get("messageId") == "") {
                    redisClient.lrem(queueDestination, 0, element);
                    element = null;
                }
            } catch(Exception e) {
                // 出现异常 把备份队列里的数据放回消费队列里
                if (!Objects.equals(ack, "ok")) {
                    redisClient.lremTransferList(element, queueDestination, queueList);
                } else {
                    if (element != null) {
                        redisClient.lrem(queueDestination, 0, element);
                    }
                }
            }
        }

//        Thread thread = new Thread(() -> {
//            System.out.println("Thread is running...");
//            List<String> item;
//            while (true) {
//                item = redisClient.blpop(key, 0);
//                System.out.println("Dequeued item: " + item);
//            }
//        });
//        thread.start();

    }

    // etcd watch 监听对象
    private Map<String, Watch.Watcher> watcherMap = new ConcurrentHashMap<>();

    /**
     * etcd watch
     */
    @Async("myAsyncThreadPool")
    public void etcdWatch() {
        System.out.println("etcdWatch is running..." + Thread.currentThread().getName());
        Watch watchClient = etcdClient.getWatchClient();
        ByteSequence key = ByteSequence.from("/config/application".getBytes());


        // 查询条件中指定只返回key
        GetOption getOption = GetOption.newBuilder().withCountOnly(true).build();
        // 如果数量小于1,表示指定的key在etcd中不存在
//        if (etcdService.getRange(key, getOption).getCount()<1) {
//            String errorDesc = String.format("[%s] not exists", key);
//            log.error(errorDesc);
//            return errorDesc + " " + new Date();
//        }

        final String watchKey = String.valueOf(key);
        // 实例化一个监听对象，当监听的key发生变化时会被调用
        Watch.Listener listener = Watch.listener(watchResponse -> {
            log.info("收到[{}]的事件", watchKey);

            // 被调用时传入的是事件集合，这里遍历每个事件
            watchResponse.getEvents().forEach(watchEvent -> {
                // 操作类型
                WatchEvent.EventType eventType = watchEvent.getEventType();
                // 操作的键值对
                KeyValue keyValue = watchEvent.getKeyValue();

                switch (eventType) {
                    case DELETE:
                        //当KEY被删除触发
                        log.info("删除事件：type={}, key={}, value={}",
                                eventType,
                                keyValue.getKey().toString(UTF_8),
                                keyValue.getValue().toString(UTF_8));
                        break;
                    case PUT:
                        //当key推送触发
                        log.info("更新事件：type={}, key={}, value={}",
                                eventType,
                                keyValue.getKey().toString(UTF_8),
                                keyValue.getValue().toString(UTF_8));

                        break;
                    default:
                        log.info("其它事件：type={}, key={}, value={}",
                                eventType,
                                keyValue.getKey().toString(UTF_8),
                                keyValue.getValue().toString(UTF_8));
                        break;
                }

                // 如果是删除操作，就把该key的Watcher找出来close掉
                if (WatchEvent.EventType.DELETE.equals(eventType)
                        && watcherMap.containsKey(watchKey)) {
                    Watch.Watcher watcher = watcherMap.remove(watchKey);
                    if (watcher != null) {
                        watcher.close();
                        log.info("关闭[{}]的事件", watchKey);
                    }
                }
            });
        });

        // 添加监听
        Watch.Watcher watcher = watchClient.watch(key, listener);
        // 将这个Watcher放入内存中保存，如果该key被删除就要将这个Watcher关闭
        watcherMap.put(watchKey, watcher);
        System.out.println("开始监听事件!");
        Scanner sc = new Scanner(System.in);
        sc.nextLine();


//        try {
//            watchClient.watch(key, new Watch.Listener() {
//                @Override
//                public void onNext(WatchResponse response) {
//                    for (WatchEvent event : response.getEvents()) {
//                        System.out.println("操作: " + event.getEventType());
//                        KeyValue kv = event.getKeyValue();
//                        System.out.println("当前kv: ");
//                        System.out.println(new String(kv.getKey().getBytes()));
//                        System.out.println(new String(kv.getValue().getBytes()));
//                        System.out.println(kv.getVersion());
//                        System.out.println(kv.getCreateRevision());
//                        System.out.println(kv.getModRevision());
//                        System.out.println(kv.getLease());
//                    }
//                }
//
//                @Override
//                public void onError(Throwable throwable) {
//                    log.info("停止监听... error: {}", throwable.getMessage());
//                }
//
//                @Override
//                public void onCompleted() {
//                    log.info("结束监听...");
//                }
//            });
//            System.out.println("键盘输入结束监听!");
//            Scanner sc = new Scanner(System.in);
//            sc.nextLine();
//        } finally {
//            if (watchClient != null) {
//                watchClient.close();
//            }
//        }

    }
}